package com.源码分析.processor;

import java.util.stream.IntStream;

import org.junit.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import reactor.core.publisher.DirectProcessor;
import reactor.core.publisher.Flux;

// 它不支持backpressure特性，如果publisher发布了N个数据，如果其中一个subscriber请求数<N，则抛出IllegalStateException.
public class DirectProcessorDemo {

	static final Logger LOGGER = LoggerFactory.getLogger(DirectProcessorDemo.class);

	@Test
	public void testDirectProcessor(){
        DirectProcessor<Integer> directProcessor = DirectProcessor.create();
        Flux<Integer> flux = directProcessor
                .filter(e -> e % 2 == 0)
                .map(e -> e +1);
        flux.subscribe(new Subscriber<Integer>() {
            private Subscription s;
            @Override
            public void onSubscribe(Subscription s) {
                this.s = s;
                s.request(3);
            }

            @Override
            public void onNext(Integer integer) {
                LOGGER.info("subscribe:{}",integer);
            }

            @Override
            public void onError(Throwable t) {
                LOGGER.error(t.getMessage(),t);
            }

            @Override
            public void onComplete() {
            	LOGGER.info("onComplete");
            }
        });

        IntStream.range(1, 20)
                .forEach(e -> {
                    directProcessor.onNext(e);
                });

        // 成功的终端状态。
        // 即使再次调用subscribe .request(long)，也不会发送其他事件
        directProcessor.onComplete();
        // 订阅此流量并无限期阻塞，直到上游指示其最后的数值或完成为止。返回该值，如果通量完成为空，则返回null。如果流量错误，则抛出originalexception(如果是checkedexception，则包装在RuntimeException中)。
        // 注意，每个blockLast()将触发一个新的订阅:换句话说，结果可能会错过来自热门出版商的信号。
        directProcessor.blockLast();
    }

}
